public async ValueTask<ConfirmsAwareChannel> GetPublishChannel(CancellationToken cancellationToken = default)
        {
            var maybePublishChannel = publishChannel;
            if (maybePublishChannel is not null)
            {
                var channel = await maybePublishChannel.Value.ConfigureAwait(false);
                if (channel.IsOpen)
                {
                    return channel;
                }

                Interlocked.CompareExchange(ref publishChannel, null, maybePublishChannel);
            }

            var maybeNewChannel = new Lazy<Task<ConfirmsAwareChannel>>(() => Task.Run(async () =>
            {
                var newChannel = new ConfirmsAwareChannel(connection!, routingTopology);
                await newChannel.Initialize(cancellationToken).ConfigureAwait(false);
                return newChannel;
            }));

            var actualLazy = Interlocked.CompareExchange(ref publishChannel, maybeNewChannel, null) ?? publishChannel!;

            return await actualLazy.Value.ConfigureAwait(false);
        }

        public async ValueTask ReturnPublishChannel(ConfirmsAwareChannel channel, CancellationToken cancellationToken = default)
        {
            if (channel.IsOpen)
            {
                return;
            }

            var maybePublishChannel = publishChannel;
            if (maybePublishChannel is not null)
            {
                var currentChannel = await maybePublishChannel.Value.ConfigureAwait(false);
                if (ReferenceEquals(currentChannel, channel))
                {
                    Interlocked.CompareExchange(ref publishChannel, null, maybePublishChannel);
                    await channel.DisposeAsync().ConfigureAwait(false);
                }
            }
        }

#pragma warning disable PS0018
        public async ValueTask DisposeAsync()
#pragma warning restore PS0018
        {
            if (disposed)
            {
                return;
            }

            await stoppingTokenSource.CancelAsync().ConfigureAwait(false);
            stoppingTokenSource.Dispose();

            var oldConnection = Interlocked.Exchange(ref connection, null);
            oldConnection?.Dispose();

            var maybeOldChannel = Interlocked.Exchange(ref publishChannel, null);
            if (maybeOldChannel is { IsValueCreated: true })
            {
                var oldChannel = await maybeOldChannel.Value.ConfigureAwait(false);
                await oldChannel.DisposeAsync().ConfigureAwait(false);
            }

            disposed = true;
        }

        readonly CancellationTokenSource stoppingTokenSource = new();
        volatile IConnection? connection;
        volatile Lazy<Task<ConfirmsAwareChannel>>? publishChannel;
        bool disposed;